// --------------------------------------------------------------------------------------------------------------------
// <copyright file="Receiver.cs" company="Microsoft Corporation">
// The MIT License (MIT)
// 
// Copyright (c) 2014, Microsoft Corporation
// 
// Permission is hereby granted, free of charge, to any person obtaining a copy
//  of this software and associated documentation files (the "Software"), to deal
//  in the Software without restriction, including without limitation the rights
//  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
//  copies of the Software, and to permit persons to whom the Software is
//  furnished to do so, subject to the following conditions:
// 
// The above copyright notice and this permission notice shall be included in
//  all copies or substantial portions of the Software.
// 
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
//  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
//  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
//  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
//  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
//  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
//  THE SOFTWARE.
// </copyright>
// --------------------------------------------------------------------------------------------------------------------
namespace Microsoft.Robotics.Runtime
{
    using System;
    using System.Collections.Concurrent;
    using System.Collections.Generic;
    using System.Diagnostics;
    using System.Threading;
    using Microsoft.Robotics.PerformanceCounters;

    /// <summary>
    /// A simple forwarder/receiver that stores only the last message.
    /// </summary>
    /// <typeparam name="TMessage">The message type to convert to and deliver (might be different than the message type passed in).</typeparam>
    [System.Diagnostics.CodeAnalysis.SuppressMessage(
        "Microsoft.Design",
        "CA1001:TypesThatOwnDisposableFieldsShouldBeDisposable",
        Justification = "Internal and only used by PipelineElement.  There is no opportunity for disposing a reciever until the pipeline element is being garbage collected anyaway")]
    internal class Receiver<TMessage> : IReceiver
        where TMessage : AgentMessage
    {
        /// <summary>
        /// The default size of the receiver queue.
        /// </summary>
        private const int DefaultQueueSize = 3;

        /// <summary>
        /// The agent host to deliver to.
        /// </summary>
        private PipelineElement host;

        /// <summary>
        /// The target of the message, which is either a subscription or control point. 
        /// </summary>
        private Action<TMessage> destination;

        /// <summary>
        /// The queue of messages stored. 
        /// This is a queue of object rather than <typeparamref name="TMessage"/> because the queue stores the input messages, not the output ones.
        /// This way the receiving pipeline element pays the type conversion price.
        /// </summary>
        private BlockingCollection<object> queue;

        /// <summary>
        /// If false, the receiver does not accept new messages.
        /// Changes to true when the agent is activated.
        /// Changes back to false when the agent is deactivated.
        /// </summary>
        private bool isActive;

        /// <summary>
        /// If true, the receiver will block any publisher calling it until the previous message is delivered.
        /// Defaults to false, which means the oldest message is dropped (is replaced by the new one)
        /// </summary>
        private bool blockIfFull;

        /// <summary>
        /// The performance counters to use
        /// </summary>
        private Dictionary<RuntimeCounters, PerformanceCounter> counters;

        /// <summary>
        /// The performance counter instance name.
        /// </summary>
        private string counterInstanceName;

        /// <summary>
        /// Initializes a new instance of the Receiver class.
        /// </summary>
        /// <param name="host">
        /// The host to notify when a new message is available.
        /// </param>
        /// <param name="destination">
        /// The receiving point.
        /// </param>
        /// <param name="blockIfFull">
        /// If true, the receiver will block any publisher calling it until the previous message is delivered.
        /// Defaults to false, which means the oldest message is dropped (is replaced by the new one)
        /// </param>
        /// <param name="queueSize">The maximum size of the queue that holds the messages pending delivery.</param>
        public Receiver(PipelineElement host, Action<TMessage> destination, bool blockIfFull, int queueSize)
        {
            this.host = host;
            this.destination = destination;
            this.blockIfFull = blockIfFull;
            this.queue = new BlockingCollection<object>(new ConcurrentQueue<object>(), (queueSize > 0) ? queueSize : DefaultQueueSize);
        }

        /// <summary>
        /// Initializes a new instance of the Receiver class.
        /// </summary>
        /// <param name="host">
        /// The host to notify when a new message is available.
        /// </param>
        /// <param name="destination">
        /// The receiving point.
        /// </param>
        /// <param name="blockIfFull">
        /// If true, the receiver will block any publisher calling it until the previous message is delivered.
        /// Defaults to false, which means the oldest message is dropped (is replaced by the new one)
        /// </param>
        /// <param name="queueSize">The maximum size of the queue that holds the messages pending delivery.</param>
        public Receiver(PipelineElement host, ISubscriptionReceiver<TMessage> destination, bool blockIfFull, int queueSize)
            : this(host, destination.Receive, blockIfFull, queueSize)
        {
            // add the perf counters
            this.counterInstanceName = string.Format("{0}-Rec({1})", this.host.Name, typeof(TMessage).Name);
            this.counters = RuntimeCounterManager.AddInstance(this.counterInstanceName);
        }

        /// <summary>
        /// Initializes a new instance of the Receiver class.
        /// </summary>
        /// <param name="host">The host to notify when a new message is available.</param>
        /// <param name="destination">The receiving point.</param>
        /// <param name="blockIfFull">
        /// If true, the receiver will block any publisher calling it until the previous message is delivered.
        /// Defaults to false, which means the oldest message is dropped (is replaced by the new one)
        /// </param>
        /// <param name="queueSize">The maximum size of the queue that holds the messages pending delivery.</param>
        public Receiver(PipelineElement host, ISubscriptionReceiver destination, bool blockIfFull, int queueSize)
            : this(host, destination.Receive, blockIfFull, queueSize)
        {
            // add the perf counters
            this.counterInstanceName = string.Format("{0}-Rec({1})", this.host.Name, typeof(TMessage).Name);
            this.counters = RuntimeCounterManager.AddInstance(this.counterInstanceName);
        }

        /// <summary>
        /// Initializes a new instance of the <see cref="Receiver{TMessage}"/> class. 
        /// Initializes a new instance of the Receiver class.
        /// </summary>
        /// <param name="host">
        /// The host to notify when a new message is available.
        /// </param>
        /// <param name="destination">
        /// The receiving point.
        /// </param>
        /// <param name="blockIfFull">
        /// If true, the receiver will block any publisher calling it until the previous message is delivered.
        /// Defaults to false, which means the oldest message is dropped (is replaced by the new one)
        /// </param>
        /// <param name="queueSize">The maximum size of the queue that holds the messages pending delivery.</param>
        public Receiver(PipelineElement host, IControlReceiver<TMessage> destination, bool blockIfFull, int queueSize)
            : this(host, destination.ReceiveControl, blockIfFull, queueSize)
        {
            // add the perf counters
            this.counterInstanceName = string.Format("{0}-RecCtrl({1})", this.host.Name, typeof(TMessage).Name);
            this.counters = RuntimeCounterManager.AddInstance(this.counterInstanceName);
        }

        /// <summary>
        /// Initializes a new instance of the Receiver class.
        /// </summary>
        /// <param name="host">The host to notify when a new message is available.</param>
        /// <param name="destination">The receiving point.</param>
        /// <param name="blockIfFull">
        /// If true, the receiver will block any publisher calling it until the previous message is delivered.
        /// Defaults to false, which means the oldest message is dropped (is replaced by the new one)
        /// </param>
        /// <param name="queueSize">The maximum size of the queue that holds the messages pending delivery.</param>
        public Receiver(PipelineElement host, IControlReceiver destination, bool blockIfFull, int queueSize)
            : this(host, destination.ReceiveControl, blockIfFull, queueSize)
        {
            // add the perf counters
            this.counterInstanceName = string.Format("{0}-RecCtrl({1})", this.host.Name, typeof(TMessage).Name);
            this.counters = RuntimeCounterManager.AddInstance(this.counterInstanceName);
        }

        /// <summary>
        /// Initializes a new instance of the <see cref="Receiver{TMessage}"/> class. 
        /// This constructor is used by proxies.
        /// </summary>
        /// <param name="host">
        /// The host to notify when a new message is available.
        /// </param>
        /// <param name="destination">
        /// A forwarder that will receive the message.
        /// </param>
        /// <param name="blockIfFull">
        /// If true, the receiver will block any publisher calling it until the previous message is delivered.
        /// Defaults to false, which means the oldest message is dropped (is replaced by the new one)
        /// </param>
        /// <param name="queueSize">The maximum size of the queue that holds the messages pending delivery.</param>
        public Receiver(PipelineElement host, IForwarder destination, bool blockIfFull, int queueSize)
            : this(host, message => destination.Post(message), blockIfFull, queueSize)
        {
        }

        /// <summary>
        /// Gets or sets a value indicating whether the receiver accepts new messages or not
        /// </summary>
        public bool IsActive
        {
            get { return this.isActive; }
            set { this.isActive = value; }
        }

        /// <summary>
        /// Accepts and stores a message for delivery. The message might not be of the exact type yet (but needs to implement a compatible datacontract). 
        /// If needed (and only then), type conversion happens when the message is consumed.
        /// </summary>
        /// <param name="message">The message to store.</param>
        /// <param name="blockUntilDelivered">
        /// If true, the call blocks if there is already a message pending delivery. 
        /// The default behavior is to replace the message pending delivery.
        /// </param>
        public virtual void Post(object message, bool blockUntilDelivered = false)
        {
            object previousMessage = null;
            if (!this.isActive)
            {
                return;
            }

            // if we are supposed to wait, then wait regardless of the queue size, but only if the host is active
            if (this.host.IsActive && (blockUntilDelivered || this.blockIfFull))
            {
                this.queue.Add(message);
            }
            else
            {
                // If there's no room in the queue, remove the oldest message. 
                // Note: we might need to develop the prunning logic further, to detect the cases where a queue is permanently backed up
                // If that happens, the agent is "living in the past", suffering from a permanent message delay. 
                // It would be better to dump more messages instead, to keep the agent anchored in the present
                // The msg delay perf counter is a great indicator of whether this situation occurs (and as of Jan 201 this doesn't appear to be an issue).
                if (!this.queue.TryAdd(message))
                {   
                    // a race condition is possible here: we could remove the second oldest message (because the consumer just removed the oldest one after we checked)
                    // not a big deal though, since the queue is already full, and the consumer should expect and be resilient to message loss.
                    this.queue.TryTake(out previousMessage);

                    // enqueue the message, there is room now
                    this.queue.Add(message);

                    if (this.counters != null)
                    {
                        this.counters[RuntimeCounters.DroppedMessages].Increment();
                    }
                }

                if (this.counters != null)
                {
                    this.counters[RuntimeCounters.QueueSize].RawValue = this.queue.Count;
                }
            }

            if (this.queue.Count > 0)
            {
                // Another race condition possible here: we wake up the pipeline element, which checks for messages, finds none and goes back to sleep. 
                // Not an issue, no real harm.
                this.host.NotifyMessageReady();
            }
        }

        /// <summary>
        /// Called on the agent host thread to deliver the message to the receiving point.
        /// </summary>
        /// <returns>True if a message was delivered, false otherwise</returns>
        public bool TryDeliver()
        {
            object message = null;
            if (!this.queue.TryTake(out message))
            {
                return false;
            }
            
            Stopwatch sw = Stopwatch.StartNew();
            TMessage outputMessage = this.Convert(message);
            long delayTime = this.host.GetPipelineTime() - outputMessage.Timestamp;
            this.destination(outputMessage);

            sw.Stop();
            if (this.counters != null)
            {
                this.counters[RuntimeCounters.ProcessedMessages].Increment();
                this.counters[RuntimeCounters.ProcessingTime].RawValue = sw.ElapsedMilliseconds;
                this.counters[RuntimeCounters.DelayTime].RawValue = delayTime;
            }

            return true;
        }

        /// <summary>
        /// Converts an input message to the right output type (assuming they have the same datacontract).
        /// </summary>
        /// <param name="inputMessage">The input message.</param>
        /// <returns>The output message.</returns>
        protected virtual TMessage Convert(object inputMessage)
        {
            // convert the message to the correct type
            if (!(inputMessage is TMessage))
            {
                // [TODO]: convert based on DataContract 
                throw new NotImplementedException("DataContract conversion NYI.");
            }

            return (TMessage)inputMessage;
        }
    }
}
